package service;

import model.command.Command;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import utils.LoggerUtil;
import utils.Role;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

public class ClusterSyncService {
    private final Store store;
    private final Node selfNode;
    private final List<Node> clusterNodes;
    private final Map<String, ObjectOutputStream> replicaConnections = new ConcurrentHashMap<>();
    private final ExecutorService executor = Executors.newCachedThreadPool();
    private final Logger LOGGER = LoggerFactory.getLogger(ClusterSyncService.class);

    //异步队列
    private final BlockingQueue<Command> syncQueue = new LinkedBlockingQueue<>();
    private ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    public ClusterSyncService(Store store, Node selfNode, List<Node> clusterNodes) {
        this.store = store;
        this.selfNode = selfNode;
        this.clusterNodes = clusterNodes;

        startAsyncSyncTask();
    }

    public void start() {
        if (selfNode.getRole() == Role.MASTER) {
            startSyncServer();
        } else {
            connectToMaster();
        }
    }

    private void startSyncServer() {
        executor.submit(() -> {
            try (ServerSocket serverSocket = new ServerSocket(getSyncPort())) {
                LoggerUtil.info(LOGGER, "Master sync server started on port: " + getSyncPort());

                while (true) {
                    Socket socket = serverSocket.accept();
                    LoggerUtil.info(LOGGER, "New slave connected: " + socket.getRemoteSocketAddress());

                    // 获取从节点的输入输出流
                    ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
                    ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());

                    // 读取从节点的 nodeId
                    String slaveNodeId = ois.readUTF(); //读取从节点 ID
                    LoggerUtil.info(LOGGER, "Received connection from slave node: {}", slaveNodeId);

                    // 关闭旧连接（如有）
                    ObjectOutputStream existingOos = replicaConnections.put(slaveNodeId, oos);
                    if (existingOos != null) {
                        existingOos.close(); //关闭旧连接
                        LoggerUtil.info(LOGGER, "Closed old connection for slave node: {}", slaveNodeId);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    private void connectToMaster() {
        Node masterNode = clusterNodes.stream()
                .filter(n -> n.getRole() == Role.MASTER)
                .findFirst()
                .orElseThrow();

        executor.execute(() -> {
            try {
                String[] addrParts = masterNode.getAddress().split(":");
                String masterHost = addrParts[0];
                int masterPort = Integer.parseInt(addrParts[1]) + 1000;

                Socket socket = new Socket(masterHost, masterPort);
                ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());

                ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
                oos.writeUTF(selfNode.getNodeId()); // 发送自身 ID
                oos.flush();

                while (true) {
                    Command command = (Command) ois.readObject();
                    ((NormalStore) store).applyCommand(command);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    public void asyncSyncCommand(Command command) {
        if (selfNode.getRole() == Role.MASTER) {
            syncQueue.offer(command);
        }
    }

    private void startAsyncSyncTask() {
        scheduler.scheduleAtFixedRate(() -> {
            List<Command> commands = new ArrayList<>();
            // 每次最多同步 1000 条命令
            syncQueue.drainTo(commands, 1000);

            if (commands.isEmpty()) return;

            List<String> failedNodes = new ArrayList<>();

            for (Map.Entry<String, ObjectOutputStream> entry : replicaConnections.entrySet()) {
                try {
                    for (Command cmd : commands) {
                        entry.getValue().writeObject(cmd);
                    }
                    entry.getValue().flush();
                } catch (IOException e) {
                    failedNodes.add(entry.getKey());
                }
            }

            // 清理失效连接
            for (String nodeId : failedNodes) {
                replicaConnections.remove(nodeId);
                LoggerUtil.info(LOGGER, "Removed failed connection to slave node: {}", nodeId);
            }

        }, 30, 30, TimeUnit.SECONDS);
        // 每 30 秒同步一次
    }


    private int getSyncPort() {
        return Integer.parseInt(selfNode.getAddress().split(":")[1]) + 1000;
    }

    private int getSyncPort(Node node) {
        return Integer.parseInt(node.getAddress().split(":")[1]) + 1000;
    }

    public Node getSelfNode() {
        return selfNode;
    }

    public void shutdown() {
        scheduler.shutdownNow();
        try {
            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            scheduler.shutdownNow();
        }

        // 关闭所有连接
        replicaConnections.values().forEach(oos -> {
            try {
                oos.close();
            } catch (IOException e) {
                LoggerUtil.info(LOGGER, "Failed to close ObjectOutputStream: {}", e.getMessage());
            }
        });
        replicaConnections.clear();
    }

}
